﻿#!perl
package TaskMgr;
use strict;
use warnings;
use threads;
use threads::shared;
use Worker;
use DataMgr;

my $instance = undef;
my $queenSignal : shared = 0;

sub getInstance : locked method{
	return $instance if $instance;
	my $class = shift;
	my %self : shared = (
		maxThreads => 1,	
		taskQueenLen => 100,
		maxStatusQueenLen => 2,
		maxResultQueenLen => 2,
		maxNewTaskQueenLen => 2,
		@_
	);
	$self{taskQueen} = &share([]);
	$self{statusQueen} = &share([]);
	$self{newTaskQueen} = &share([]);
	$self{resultQueen} = &share([]);
	$instance = bless \%self, $class;
	return $instance;
}

sub addNewTasks : locked method {
	my ($self, $tasks) = @_;
	lock $self;
	my $queen = $self->{newTaskQueen};
	push @{$queen}, @{$tasks};
	if ($#{$queen} >= $self->{maxNewTaskQueenLen} - 1) {
		lock $queenSignal;
		$queenSignal = 'newTaskQueenFull';
		#print "addNewTasks:cond_signal $queenSignal\n";
		cond_signal $queenSignal;
		my $timeout = time + 5;
		#print "addNewTasks:cond_wait $queenSignal\n";
		print "addNewTasks: timeout\n" unless cond_timedwait $queenSignal, $timeout;	
	}
}

sub getATask : locked method {
	my $self = shift;
	lock $self;
	my $queen = $self->{taskQueen};
	if ($#{$queen} == -1) {
		lock $queenSignal;
		$queenSignal = 'taskQueenEmpty';
		#print "getATask:cond_signal $queenSignal\n";
		cond_signal $queenSignal;
		my $timeout = time + 5;
		#print "getATask:cond_wait $queenSignal\n";
		print "getATask: timeout\n" unless cond_timedwait $queenSignal, $timeout;
	}
	return shift @{$queen} if $#{$queen} != -1;
	return 0;
}

sub setTaskStatus : locked method {
	my ($self, $status) = @_;
	lock $self;
	my $queen = $self->{statusQueen};
	push @{$queen}, $status;
	if ($#{$queen} + 1 >= $self->{maxStatusQueenLen}) {
		lock $queenSignal;
		$queenSignal = 'statusQueenFull';
		#print "setTaskStatus:cond_signal $queenSignal\n";
		cond_signal $queenSignal;
		my $timeout = time + 5;
		#print "setTaskStatus:cond_wait $queenSignal\n";
		print "setTaskStatus: timeout\n" unless cond_timedwait $queenSignal, $timeout;
	}
}

sub setResult : locked method {
	my ($self, $result) = @_;
	lock $self;
	my $queen = $self->{resultQueen};
	push @{$queen}, @{$result};
	if ($#{$queen} + 1 >= $self->{maxResultQueenLen}) {
		lock $queenSignal;
		$queenSignal = 'resultQueenFull';
		#print "setResult:cond_signal $queenSignal\n";
		cond_signal $queenSignal;
		my $timeout = time + 5;
		#print "setResult:cond_wait $queenSignal\n";
		print "setResults: timeout\n" unless cond_timedwait $queenSignal, $timeout;
	}
}

my $workThread = sub {
	my $worker = Worker->new();
	my $taskmgr = shift;
	my $tid = threads->self->tid;
	my $sleepTime = 0.1;
	my $maxSleepTime = 10;
	$SIG{'KILL'} = sub {
		print "WorkThread$tid exit...\n";
		threads->exit;	
	}
	print "Create WorkThread.$tid\n";
	while (1) {
		my $task = getATask $taskmgr;
		if (not $task) {
			print "WorkThread$tid: get no task!\n";
			sleep $sleepTime;
			$sleepTime *= 2 if $sleepTime < $maxSleepTime;
		} else {
			$sleepTime = 0.1;
		}
		next unless $task;
		print "WorkThread$tid fetch task: $task \n";
		my $result = $worker->do($task);
		my $status;
		if ($result) {
			$status = $worker->success($task);	
		} else {
			$status = $worker->fail($task);
		}
		$taskmgr->setTaskStatus($status);
		$taskmgr->addNewTasks($result->{newTasks}) if $result and $result->{newTasks};
		$taskmgr->setResult($result->{result}) if $result and $result->{result};
	}
};

my $queenAdminThread = sub {
	$SIG{'KILL'} = sub {
		print "queenAdminThread exit...\n";
		threads->exit;	
	}
	my $taskmgr = shift;
	my $datamgr = DataMgr->new;
	{
		lock $queenSignal;
		cond_signal $queenSignal; #tell the main thread inited ok
	}
	lock $queenSignal;
	while (1) {
		until ($queenSignal) {
			#print "queenAdminThread:cond_wait $queenSignal\n";
			cond_wait $queenSignal;
			CASE : {
				$queenSignal eq 'taskQueenEmpty' and do {
					my $queen = $taskmgr->{taskQueen};
					my $wantTotal = $taskmgr->{taskQueenLen} - scalar @{$queen};
					if ($wantTotal > 0) {
						my $datas = $datamgr->getTasks($wantTotal);
						push @{$queen}, @{$datas};
					}
					last CASE;	
				};
				$queenSignal eq 'statusQueenFull' and do {
					my $queen = $taskmgr->{statusQueen};
					$datamgr->setStatus($queen);
					$#{$queen} = -1;
					pop @{$queen}; #a bug about the shared array that leave one element exist
					last CASE;
				};
				$queenSignal eq 'newTaskQueenFull' and do {
					my $queen = $taskmgr->{newTaskQueen};
					$datamgr->setNewTasks($queen);
					$#{$queen} = -1;
					pop @{$queen}; #a bug about the shared array that leave one element exist
					last CASE;
				};
				$queenSignal eq 'resultQueenFull' and do {
					my $queen = $taskmgr->{resultQueen};
					$datamgr->setResults($queen);
					$#{$queen} = -1;
					pop @{$queen}; #a bug about the shared array that leave one element exist
					last CASE;
				};
			}
			unless ($queenSignal) {
				print "queenAdminThread:cond_signal again!\n";
			  	cond_signal $queenSignal;	
			}
		}
		
		$queenSignal = 0;
		#print "queenAdminThread:cond_signal $queenSignal\n";
		cond_signal $queenSignal;
	}
};

sub run {
	my $self = shift;
	my @threads;
	{
		lock $queenSignal;
		push @threads, threads->new($queenAdminThread, $self);
		cond_wait $queenSignal;  #wait queenAdminThread init;
	}
	sleep 0.1;
		
	push @threads, threads->new($workThread, $self) for 1..$self->{maxThreads};
	$SIG{'INT'} = sub {
		$_->kill('KILL')->join foreach @threads;	
	}
	while (1) {
		sleep 1000;	
	}
}

1;
